package com.youxin.dataStream;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;

import java.util.Properties;

/**
 * 对接kafka数据源
 */
public class StreamKafkaSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment evn = StreamExecutionEnvironment.getExecutionEnvironment();

        //kafka的相关的配置
        Properties properties = new Properties();
        properties.put("bootstrap.servers","hadoop-1:9092");
        properties.put("zookeeper.connect","hadoop-1:2181");
        properties.put("group.id","flink");
        properties.put("auto.offset.reset","latest");
        properties.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

        DataStreamSource<String> topicDataSource = evn.addSource(new FlinkKafkaConsumer010<String>("topic", new SimpleStringSchema(), properties));
        /**
         * you can use map flatMap and so on
         */
        topicDataSource.addSink(new PrintSinkFunction<>());

        evn.execute("flink kafka job");


    }
}
